package com.sunday.common.mq.kafka.study.spring.e11_Serialization_Deserialization_Message_Conversion;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.kafka.DefaultKafkaConsumerFactoryCustomizer;
import org.springframework.boot.autoconfigure.kafka.DefaultKafkaProducerFactoryCustomizer;
import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.KafkaListenerErrorHandler;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;

import java.util.Map;

/**
 * https://docs.spring.io/spring-kafka/docs/current/reference/html/#json-serde
 * <p>
 * https://docs.spring.io/spring-kafka/docs/current/reference/html/#serdes-json-config
 * Configuration Properties
 * JsonSerializer.ADD_TYPE_INFO_HEADERS (default true): You can set it to false to disable this feature on the JsonSerializer (sets the addTypeInfo property).
 * JsonSerializer.TYPE_MAPPINGS (default empty): See Mapping Types.
 * JsonDeserializer.USE_TYPE_INFO_HEADERS (default true): You can set it to false to ignore headers set by the serializer.
 * JsonDeserializer.REMOVE_TYPE_INFO_HEADERS (default true): You can set it to false to retain headers set by the serializer.
 * JsonDeserializer.KEY_DEFAULT_TYPE: Fallback type for deserialization of keys if no header information is present.
 * JsonDeserializer.VALUE_DEFAULT_TYPE: Fallback type for deserialization of values if no header information is present.
 * JsonDeserializer.TRUSTED_PACKAGES (default java.util, java.lang): Comma-delimited list of package patterns allowed for deserialization. * means deserialize all.
 * JsonDeserializer.TYPE_MAPPINGS (default empty): See Mapping Types.
 * JsonDeserializer.KEY_TYPE_METHOD (default empty): See Using Methods to Determine Types.
 * JsonDeserializer.VALUE_TYPE_METHOD (default empty): See Using Methods to Determine Types.
 */
@Slf4j
@SpringBootApplication
public class JsonSerializerApplication {

    public static void main(String[] args) {
        SpringApplication.run(JsonSerializerApplication.class, args);
    }

    @Bean
    public DefaultKafkaConsumerFactory<?, ?> kafkaConsumerFactory(KafkaConnectionDetails connectionDetails,
                                                                  ObjectProvider<DefaultKafkaConsumerFactoryCustomizer> customizers,
                                                                  KafkaProperties prop
    ) {
        Map<String, Object> properties = prop.buildConsumerProperties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getProducerBootstrapServers());


        /**
         * 方案 1
         */
//        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class.getName());
//        properties.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
//        DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory<>(properties);

        /**
         * 方案 2
         * https://docs.spring.io/spring-kafka/docs/current/reference/html/#serdes-type-methods
         */
        DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory<>(
                properties,
                new StringDeserializer(),
                new JsonDeserializer().trustedPackages("*")
        );

        customizers.orderedStream().forEach((customizer) -> customizer.customize(factory));
        return factory;
    }

    @Bean
    public DefaultKafkaProducerFactory<?, ?> kafkaProducerFactory(KafkaConnectionDetails connectionDetails,
                                                                  ObjectProvider<DefaultKafkaProducerFactoryCustomizer> customizers,
                                                                  KafkaProperties prop
    ) {
        Map<String, Object> properties = prop.buildProducerProperties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getProducerBootstrapServers());

        /**
         * 方案 1
         */
//        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName());
//        DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory<>(properties);

        /**
         * 方案 2
         */
        DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory<>(
                properties,
                new StringSerializer(),
                new JsonSerializer()
        );

        String transactionIdPrefix = prop.getProducer().getTransactionIdPrefix();
        if (transactionIdPrefix != null) {
            factory.setTransactionIdPrefix(transactionIdPrefix);
        }
        customizers.orderedStream().forEach((customizer) -> customizer.customize(factory));
        return factory;
    }

    /**
     * https://docs.spring.io/spring-kafka/docs/current/reference/html/#kafka-validation
     */
    @Bean
    public KafkaListenerErrorHandler validationErrorHandler() {
        return (m, e) -> {
            log.error("{} - {}", m, e.toString());
            return null;
        };
    }

    @Bean
    public NewTopic topic7() {
        return TopicBuilder.name("topic7")
                .partitions(3)
                .replicas(1)
                .build();
    }

}
